package com.seventeen.monster.redisson.core;

import com.seventeen.monster.common.utils.spring.SpringUtils;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author seventeen
 */
public class DelayedQueueUtils {
    private final static Logger log = LoggerFactory.getLogger(DelayedQueueUtils.class);

    private static RedissonClient redissonClient;

    private static RedissonClient getRedissonClient() {
        if (redissonClient == null) {
            redissonClient = SpringUtils.getBean(RedissonClient.class);
        }
        return redissonClient;
    }

    /**
     * 添加队列
     *
     * @param queueName 队列名称
     * @param payload   消息体
     * @param delay     时间数量
     * @param timeUnit  时间单位
     * @param <T>       泛型
     */
    public static <T> void publish(String queueName, T payload, long delay, TimeUnit timeUnit) {
        Message<T> message = buildMessageView(payload);
        publish(queueName, message, delay, timeUnit);
    }

    public static <T> void publish(String queueName,
                                   T payload,
                                   Map<String, String> headers,
                                   long delay,
                                   TimeUnit timeUnit) {
        Message<T> message = buildMessageView(payload, headers);
        publish(queueName, message, delay, timeUnit);
    }

    public static <T> void publish(String queueName, Message<T> message, long delay, TimeUnit timeUnit) {
        RDelayedQueue<Message<T>> delayedQueue = getDelayedQueue(queueName);
        log.info("redisson delay queue publish, queue {} delay {} timeUnit {} msg {}",
                queueName,
                delay,
                timeUnit,
                message);
        delayedQueue.offer(message, delay, timeUnit);
    }

    private static <T> Message<T> buildMessageView(T payload) {
        Message<T> message = new Message<>();
        message.setPayload(payload);
        return message;
    }

    private static <T> Message<T> buildMessageView(T payload, Map<String, String> headers) {
        Message<T> message = new Message<>();
        message.setPayload(payload);
        message.setHeaders(headers);
        return message;
    }

    public static <T> boolean contains(String queueName, T payload) {
        RDelayedQueue<Message<T>> delayedQueue = getDelayedQueue(queueName);
        Message<T> message = buildMessageView(payload);
        return delayedQueue.contains(message);
    }

    /**
     * 重试消息发送
     *
     * @param queueName  队列名
     * @param message    消息体
     * @param retryLevel 重试级别
     * @param <T>        泛型
     */
    public static <T> void retry(String queueName, Message<T> message, int retryLevel) {
        TimeUnit timeUnit;
        long delay;
        // 策略仿照RocketMQ
        // [1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h]
        switch (retryLevel) {
            case 1:
                delay = 1;
                timeUnit = TimeUnit.SECONDS;
                break;
            case 2:
                delay = 5;
                timeUnit = TimeUnit.SECONDS;
                break;
            case 3:
                delay = 10;
                timeUnit = TimeUnit.SECONDS;
                break;
            case 4:
                delay = 30;
                timeUnit = TimeUnit.SECONDS;
                break;
            case 5:
                delay = 1;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 6:
                delay = 2;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 7:
                delay = 3;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 8:
                delay = 4;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 9:
                delay = 5;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 10:
                delay = 6;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 11:
                delay = 7;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 12:
                delay = 8;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 13:
                delay = 9;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 14:
                delay = 10;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 15:
                delay = 20;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 16:
                delay = 30;
                timeUnit = TimeUnit.MINUTES;
                break;
            case 17:
                delay = 1;
                timeUnit = TimeUnit.HOURS;
                break;
            default:
                delay = 2;
                timeUnit = TimeUnit.HOURS;
        }
        publish(queueName, message, delay, timeUnit);
    }

    private static <T> RDelayedQueue<Message<T>> getDelayedQueue(String queueName) {
        RedissonClient redissonClient = getRedissonClient();
        try {
            RQueue<Message<T>> queue = redissonClient.getQueue(queueName);
            return redissonClient.getDelayedQueue(queue);
        } catch (Exception e) {
            log.error("get delay queue fail,", e);
            throw new RuntimeException("获取延迟队列失败", e);
        }
    }
}
